package cn.itcast.flink.start;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

import static org.apache.flink.table.api.Expressions.*;

/**
 * 基于Flink Table Api实现批处理：加载文本文件数据，进行电影评分统计分析
 *
 * @author lilulu
 * @date 2023-04-10 16:39
 */
public class FlinkTableApiDemo {
    public static void main(String[] args) {
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inBatchMode().useBlinkPlanner().build();
        TableEnvironment tableEnv = TableEnvironment.create(settings);

        tableEnv.executeSql(
                "CREATE TABLE tbl_ratings(\n" +
                        "  user_id STRING,\n" +
                        "  movie_id STRING,\n" +
                        "  rating DOUBLE,\n" +
                        "  ts BIGINT\n" +
                        ") WITH (\n" +
                        "  'connector' = 'filesystem', \n" +
                        "  'path' = 'datas/ratings.data', \n" +
                        "  'format' = 'csv',\n" +
                        "  'csv.field-delimiter' = '\\t',\n" +
                        "  'csv.ignore-parse-errors' = 'true'\n" +
                        ")"
        );

        Table resultTable = tableEnv.from("tbl_ratings")
                .groupBy($("movie_id"))
                .select(
                        $("movie_id"),
                        $("movie_id").count().as("rating_people"),
                        $("rating").avg().round(2).as("rating_number")
                )
                .where(
                        $("rating_people").isGreater(400))
                .orderBy($("rating_number").desc(), $("rating_people").desc())
                .limit(10);

        resultTable.execute().print();
    }
}
